package main

import (
	"fmt"
	"github.com/Shopify/sarama"
	"log"
)

/*
同步推送,异步推送
*/
func main() {
	PushMsg()
}

// 广播推送
func PushMsg() {
	// 创建一个config
	addr := []string{"192.168.72.133:9092"}
	config := sarama.NewConfig()
	config.Producer.Return.Successes = true // 同步模式
	config.Version = sarama.V2_1_0_0

	//config.Producer.RequiredAcks = sarama.WaitForAll
	//config.Producer.Partitioner = sarama.NewRandomPartitioner
	//config.Producer.Return.Errors = true

	client, err := sarama.NewClient(addr, config)
	if err != nil {
		fmt.Println("创建Client失败：", err)
	}

	// 创建一个生产者
	producer1, err := sarama.NewSyncProducerFromClient(client)
	if err != nil {
		fmt.Println("创建producer失败：", err)
	}

	defer func() {
		if err := producer1.Close(); err != nil {
			fmt.Println(err)
		}
	}()

	// 创建一条消息
	text := "hello kate"
	// producer把消息发给kafka之后会等待结果返回
	partition, offset, err := producer1.SendMessage(&sarama.ProducerMessage{Topic: "test", Key: nil, Value: sarama.StringEncoder(text)})
	if err != nil {
		log.Fatalf("unable to produce message: %q", err)
	} else {
		fmt.Println(partition, offset)
	}

}
